package cn.doitedu.demo7;

import cn.doitedu.beans.UserAction;
import cn.doitedu.common.utils.CommonUtils;
import cn.doitedu.common.utils.StateDescriptors;
import groovy.lang.GroovyClassLoader;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;
import java.util.Map;


@Slf4j
public class CoreProcessFunction_Final extends KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String> {

    HashMap<String, RuleCalculator> ruleCalculatorPool;

    // 最近2分钟的用户行为明细缓存
    ListState<UserAction> recentActions;

    // 用户存放已处理过数据的运算机的规则id
    MapState<String, String> noNewCalculatorIds;

    GroovyClassLoader groovyClassLoader;

    Object lock;

    @Override
    public void open(Configuration parameters) throws Exception {

        lock = new Object();

        ruleCalculatorPool = new HashMap<>();

        ListStateDescriptor<UserAction> desc = new ListStateDescriptor<>("recent_actions", UserAction.class);
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.minutes(2))
                .neverReturnExpired()
                .build();
        desc.enableTimeToLive(ttlConfig);

        recentActions = getRuntimeContext().getListState(desc);


        this.noNewCalculatorIds = getRuntimeContext().getMapState(new MapStateDescriptor<String, String>("no_new_calculator_ids", String.class, String.class));


        this.groovyClassLoader = new GroovyClassLoader();


    }

    @Override
    public void processElement(UserAction currentAction, KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {

        synchronized (lock) {
            // 人为造故障
            /*if (currentAction.getEvent_id().equals("ERROR") && RandomUtils.nextInt(1, 10) % 3 == 0) {
                throw new RuntimeException("我故意崩溃了.......................你难受不难受...................");
            }*/

            ReadOnlyBroadcastState<String, RuleMeta> broadcastState = ctx.getBroadcastState(StateDescriptors.RULEMETA_BC_STATE_DES);

            /* *
             * 安插一段逻辑：
             *   判断系统是不是刚刚被failover重启
             *   如果是，则要走这段安插逻辑：恢复出崩溃之前的运算机池
             */
            CommonUtils.restoreCalculatorPool(groovyClassLoader, ruleCalculatorPool, broadcastState, getRuntimeContext());


            /* *
             * 正常逻辑
             */
            // 先把本次的用户行为放入缓存状态
            recentActions.add(currentAction);

            // 遍历运算机池
            for (Map.Entry<String, RuleCalculator> entry : ruleCalculatorPool.entrySet()) {

                // 取到一个规则运算机对象
                String ruleId = entry.getKey();
                RuleCalculator ruleCalculator = entry.getValue();

                // 判断该运算机是否是一个新上线的运算机 且是那种需要统计历史段数据的运算机
                if (!noNewCalculatorIds.contains(ruleId)   && ruleCalculator.isNeedProcessPastData()) {
                    // 如果是，则要把缓存状态中的最近2分钟历史数据交给这个运算机处理
                    for (UserAction action : recentActions.get()) {
                        ruleCalculator.calc(action, out);
                    }

                    // 然后将这个新运算加入 “老运算机id集合”
                    noNewCalculatorIds.put(ruleId, "");

                } else {
                    // 如果否,那就把本次收到的这一条行为交给它处理
                    ruleCalculator.calc(currentAction, out);
                }
            }
        }

    }

    @Override
    public void processBroadcastElement(RuleMeta ruleMeta, KeyedBroadcastProcessFunction<Long, UserAction, RuleMeta, String>.Context ctx, Collector<String> out) throws Exception {

        synchronized (lock) {
            BroadcastState<String, RuleMeta> broadcastState = ctx.getBroadcastState(StateDescriptors.RULEMETA_BC_STATE_DES);

            /* *
             * 安插一段逻辑：
             *   判断系统是不是刚刚被failover重启
             *   如果是，则要走这段安插逻辑：恢复出崩溃之前的运算机池
             */
            CommonUtils.restoreCalculatorPool(groovyClassLoader, ruleCalculatorPool, broadcastState, getRuntimeContext());


            /* *
             * 正常逻辑
             */
            String op = ruleMeta.getOp();

            // 如果是元数据删除操作，或 元数据更新操作且更新后的规则状态为0 ，都表示要下线规则
            if (op.equals("d") || (op.equals("u") && ruleMeta.getRule_status() == 0)) {

                // 从运算机池，移除该规则的运算机对象
                ruleCalculatorPool.remove(ruleMeta.getRule_id());
                log.warn("下线了一个规则运算机,rule_id:{}", ruleMeta.getRule_id());

                // 从元数据广播状态中也移除该规则的元数据
                broadcastState.remove(ruleMeta.getRule_id());

            }
            // 如果是初始读或新增了一条规则元数据，且规则的状态为1，表示要上线规则
            else if ((op.equals("r") || op.equals("c") || op.equals("u")) && ruleMeta.getRule_status() == 1) {

                // 从规则元数据中，取出规则对应的模型运算机groovy代码
                String modelCode = ruleMeta.getModel_code();
                Class<?> aClass = groovyClassLoader.parseClass(modelCode);

                RuleCalculator ruleCalculator = (RuleCalculator) aClass.newInstance();

                // 并利用元数据中的 规则参数，初始化运算机，得到具体规则运算机对象
                ruleCalculator.init(getRuntimeContext(), ruleMeta);

                // 将初始化好的运算机对象，放入运算机池
                ruleCalculatorPool.put(ruleMeta.getRule_id(), ruleCalculator);
                log.info("上线了一个规则运算机,rule_id:{}", ruleMeta.getRule_id());

                // 把该规则的元数据放入广播状态保存下来
                broadcastState.put(ruleMeta.getRule_id(), ruleMeta);

            } else {
                log.info("收到一个规则元数据，但我什么也没做,rule_id:{}", ruleMeta.getRule_id());
            }
        }
    }

}
